package iot.hbase.codec;

import com.datastax.driver.core.utils.UUIDs;
import hbase.template.CellMapper;
import hbase.template.RowMapper;
import hbase.template.RowSerializer;
import iot.entity.Location;
import iot.utils.DataPackUtil;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.UUID;

/**
 * Created by dave on 18-7-21 上午1:11.
 */
public class LocationCodec implements RowSerializer<Location>, RowMapper<Location> {

    @Override
    public List<KeyValue> serialize(Location record) {
        List<KeyValue> kv = new ArrayList<>();
        byte[] key = generateRowKey(record);
        byte[] family = "l".getBytes();
        kv.add(new KeyValue(key, family, "si".getBytes(), record.getSim().getBytes()));
        kv.add(new KeyValue(key, family, "af".getBytes(), Bytes.toBytes(record.getAlarmFlag())));
        kv.add(new KeyValue(key, family, "fl".getBytes(), Bytes.toBytes(record.getFlag())));
        kv.add(new KeyValue(key, family, "lo".getBytes(), Bytes.toBytes(record.isLocated())));
        kv.add(new KeyValue(key, family, "la".getBytes(), Bytes.toBytes(record.getLat())));
        kv.add(new KeyValue(key, family, "ln".getBytes(), Bytes.toBytes(record.getLng())));
        kv.add(new KeyValue(key, family, "he".getBytes(), Bytes.toBytes(record.getHeight())));
        kv.add(new KeyValue(key, family, "sp".getBytes(), Bytes.toBytes(record.getSpeed())));
        kv.add(new KeyValue(key, family, "di".getBytes(), Bytes.toBytes(record.getDirection())));
        kv.add(new KeyValue(key, family, "tm".getBytes(), Bytes.toBytes(record.getTime().getTime())));
        return kv;
    }

    @Override
    public byte[] generateRowKey(Location record) {
        return makeRowKey(record.getSim(), record.getUuid());
    }

    @Override
    public Location mapRow(Result result) {
        Location l = new Location();
        l.setSim(CellMapper.toString(result, "l", "si"));
        l.setAlarmFlag(CellMapper.toInt(result, "l", "af"));
        l.setFlag(CellMapper.toInt(result, "l", "fl"));
        l.setLocated(CellMapper.toBoolean(result, "l", "lo"));
        l.setLat(CellMapper.toDouble(result, "l", "la"));
        l.setLng(CellMapper.toDouble(result, "l", "ln"));
        l.setHeight(CellMapper.toInt(result, "l", "he"));
        l.setSpeed(CellMapper.toInt(result, "l", "sp"));
        l.setDirection(CellMapper.toInt(result, "l", "di"));
        l.setTime(new Date(CellMapper.toLong(result, "l", "tm")));
        return l;
    }

    public byte[] makeRowKey(String sim, UUID timeUuid) {
        List<Byte> list = DataPackUtil.getBCDBytes(sim);
        byte[] bytes = new byte[list.size() + 8];
        System.arraycopy(DataPackUtil.byteListToArray(list), 0, bytes, 0, list.size());
        System.arraycopy(DataPackUtil.getBytesFromUUID(timeUuid), 0, bytes, 6, 8);
        return bytes;
    }

    public byte[] makeStartRowKey(String sim, Date time) {
        UUID uuid = UUIDs.startOf(time.getTime());
        return makeRowKey(sim, uuid);
    }

    public byte[] makeEndRowKey(String sim, Date time) {
        UUID uuid = UUIDs.endOf(time.getTime());
        return makeRowKey(sim, uuid);
    }
}
